package com.ebupt.migu.music.autoTri.config;

import com.ebupt.migu.music.autoTri.Service.AsyncService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author xuhongsong
 * @version 1.0
 * @date 2020/9/9 20:23
 */
@Slf4j
@Component
public class KafkaConsumer {

    @Autowired
    private AsyncService asyncService;

    public static KafkaConsumer kafkaConsumer;
    @PostConstruct
    public void init() {
        kafkaConsumer = this;
    }

    /**
     *  消费者 6.16 门户接收素材的接口
     * @param consumerRecord
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.new-group-id}",topics = {"${spring.kafka.consumer.group-tops}"})
    public void receive(ConsumerRecord<String,String> consumerRecord){
        kafkaConsumer.asyncService.insertKafka(consumerRecord);
    }

    /**
     *  消费者 6.22 门户接收编目信息的接口REST版
     * @param consumerRecord
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.new-group-id}",topics = {"${spring.kafka.consumer.group-tops-Catalog}"})
    public void receives1(ConsumerRecord<String,String> consumerRecord){
        kafkaConsumer.asyncService.insertKafkaofCatalogInfo(consumerRecord);
    }

    /**
     *  消费者 6.23 门户接收统一曲库分发MV接口REST版
     * @param consumerRecord
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.new-group-id}",topics = {"${spring.kafka.consumer.group-tops-DeployMv}"})
    public void receives2(ConsumerRecord<String,String> consumerRecord){
        kafkaConsumer.asyncService.insertKafkaofdeployMv(consumerRecord);
    }

    /**
     *  消费者 6.21 门户接收统一曲库分发（振、全、随）产品接口REST版
     * @param consumerRecord
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.new-group-id}",topics = {"${spring.kafka.consumer.group-tops-wireless_product}"})
    public void receives3(ConsumerRecord<String,String> consumerRecord){
        kafkaConsumer.asyncService.insertWirelessProduct(consumerRecord);
    }

    /**
     *  消费者 11.7 变色龙实时效果数据展示 产品接口REST版
     * @param consumerRecord
     */
    @KafkaListener( groupId = "${spring.kafka.consumer.old-group-id}",topics = {"${spring.kafka.consumer.global-popup-real-time-data}"})
    public void receives4(ConsumerRecord<String,String> consumerRecord){
        kafkaConsumer.asyncService.insertPopupData(consumerRecord);
    }
}
